﻿#region Creditos

/* Autor: Edilberto Sánchez Forero	*/
/* Email: esanchezfo@gmail.com		*/
/* Licencia: LGPL, Software Libre	*/
/* País: Colombia, Bogotá			*/

#endregion

#region Librerias

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using ESF.General;
using ESF.General.Bulk;

#endregion


namespace ESF.Bulk
{
	public class TasksQueue : ITaskQueue
	{
		#region Constantes
		
		private const long TOTAL_MAXLENGTH = 2304000000;
		
		private const int INDIVIDUAL_LIST_MAXLENGTH = 48000;
		
		private const int DEFAULT_MAXTRYCOUNT = 5;
		
		// TODO: Derivarlo de las condiciones de la maquina
		// TODO: Pasarlo a un threadpool
		private const int MAX_WORKERS_COUNT = 500;
		private const int MIN_WORKERS_COUNT = 6;
		
		private const int CHECKCLEANCYCLEMS = 200;
		
		#endregion
		
		
		#region Miembros
		
		private List<Queue <ITask>> _Tasks;
		
		private List<WorkerQueue> _Workers;
		
		private string _Name;
		
		private int _MaxTryCount;
		private int _AvgTimeDurationMs;
		
		private long _TaskSize;
		private long _PendingTasksCount;
		
		private string _Started;
		private string _Lock;
		
		private Stopwatch _Stopwatch;
		
		private int _PreviousAvgTimeDurationMs;
		private int _PreviousAddedWorkersCount;
		
		private long _PreviousTotalTimeMs;
		
		private Thread _ThreadClean;
		
		private bool _EnableAddWorkers;
		
		#endregion
		
		
		#region Propiedades
		
		public int ActiveWorkersCount
		{
			get { return _Workers.Count (w => w.Active); }
		}
		
		#endregion
		
		
		#region Constructor
		
		public TasksQueue (string pName) 
			: this (pName, DEFAULT_MAXTRYCOUNT) {}
		
		
		public TasksQueue (string pName, int pMaxTryCount)
		{
			_Workers = new List<WorkerQueue> ();
			
			_Tasks = new List<Queue<ITask>> ();
			
			_Name = pName;
			
			_MaxTryCount = pMaxTryCount;
			
			_Started = string.Empty;
			
			_Stopwatch = new Stopwatch ();
		
			_PreviousAvgTimeDurationMs = 0;
			_PreviousTotalTimeMs = 0;
			_PreviousAddedWorkersCount = 0;
		
			_Lock = string.Empty;
			
			Stop ();
		}
		
		#endregion
		
		
		#region Manejo de Workers
		
		private void AddWorker ()
		{
			WorkerQueue worker;
			
		
			// Si no se han dado errores y el volumen de tareas es superior al de trabajadores
			if (_Workers.Count < MAX_WORKERS_COUNT)
			{	
				worker = _Workers.Find (w => !w.Active);
					
				if (worker == null)
				{
					worker = new WorkerQueue (Dequeue, ConfirmedTask, TreatException, _MaxTryCount, _Workers.Count);
					
					_Workers.Add (worker);
				}
				
				worker.Start ();
			}
		}
		
		#endregion
		
		
		#region Funciones de Control
		
		private void Start ()
		{
			if (!string.IsNullOrEmpty (_Started)) { return; }
		
			lock (_Started)
			{
				_Started = "Yes";
			
				_PendingTasksCount = 0;
				
				_Stopwatch.Reset ();
				_Stopwatch.Start ();
				
				_AvgTimeDurationMs = 0;
				
				_PreviousTotalTimeMs = 0;
				
				_EnableAddWorkers = true;
				
				Console.Write (" - Starting Queue for {0}...", _Name);
				
				AddWorkers (MIN_WORKERS_COUNT);
			
				_ThreadClean = new Thread (new ParameterizedThreadStart (Clean));
				_ThreadClean.Start ();
				
				Console.WriteLine (" started.");
			}
		}
		
		
		public void Stop ()
		{
			Parallel.ForEach <WorkerQueue> (_Workers, (w) =>
			{
				w.Stop ();
			});
			
			_Workers.Clear ();
			
			_Started = string.Empty;
			
			_Tasks.Clear ();
		}
		
		
		public void WaitToComplete ()
		{
			while (_PendingTasksCount > 0)
			{
				Thread.Sleep (_AvgTimeDurationMs);
			}
			
			Console.Write (" - Stoping Queue for {0}...", _Name);
			
			Stop ();
			
			Console.WriteLine (" completed in {1}", _Name, TextHelper.ToString (ref _Stopwatch));
		}
		
		#endregion
		
		
		#region Funciones
		
		private void AddWorkers (int pCount)
		{
			List<WorkerQueue> tmpWorkers;
			
			int tmpLimit;
			
			
			if (_Workers.Count > MAX_WORKERS_COUNT) { return; }
		
			tmpLimit = MAX_WORKERS_COUNT - _Workers.Count;
			
			if (pCount > tmpLimit) { pCount = tmpLimit; }
		
			// Despierta a los que no estan trabajando
			tmpWorkers = _Workers.FindAll (w => !w.Active);
			
			if (tmpWorkers.Count > 0)
			{
				if (pCount > tmpWorkers.Count) { tmpLimit = tmpWorkers.Count; }
				else { tmpLimit = pCount; }
			
				//Console.WriteLine (" -- Levantando a {0} workers dormidos", tmpLimit);
				Parallel.For (0, tmpLimit, (i) =>
				{
					tmpWorkers [i].Start ();
				});
				
				pCount -= tmpLimit;
			}
			
			if (_EnableAddWorkers && pCount > 0) 
			{
				//Console.WriteLine (" -- Adding {0} workers.", pCount);
				Parallel.For (0, pCount, (i) =>
				{
					AddWorker ();
				});
			}
		}
		
		
		private void TreatException (Exception pEx, int pTryCount)
		{
			// Sintoma de falta de recursos
			// TODO: Especializar manejo de excepciones para saber en el task cuando es un fallo grave o es reintentable
			
			if (pTryCount >= _MaxTryCount)
			{
				Console.WriteLine (" * Max Exception to try reached! *");
			
				Stop ();
				
				throw pEx;
			}
		}
		
		
		private void Clean (object pNotUsed)
		{
			List<WorkerQueue> tmpWorkers;
			
			int previousAvgTimeDurationMs;
			int tmpSleepCount;
			
			long previousTotalTimeMs;
			
			double previousAdv;
			
			long taskSize;
			long pendingTasksCount;
			
						
			taskSize = Interlocked.Read (ref _TaskSize);
			pendingTasksCount = Interlocked.Read (ref _PendingTasksCount);
			
			if (pendingTasksCount == 0)
			{
				return;
			}
			
			if (_AvgTimeDurationMs == 0 || _Workers.Count == 0) 
			{
				try
				{
					Thread.Sleep (CHECKCLEANCYCLEMS);
					Clean (pNotUsed); 
					return;
				}
				catch (StackOverflowException)
				{
					return;
				} 
			}
			
			
			previousAvgTimeDurationMs = _PreviousAvgTimeDurationMs;
			previousTotalTimeMs = _PreviousTotalTimeMs;
			
			previousAdv = ((taskSize - pendingTasksCount) * 100.0) / taskSize;
			
			
			_PreviousAvgTimeDurationMs = _AvgTimeDurationMs;
			
			_PreviousTotalTimeMs = (long) Math.Ceiling ((taskSize * _AvgTimeDurationMs * 2.0) / _Workers.Count);
			
			
			Console.Clear ();
			Console.WriteLine (" - {0}:", _Name);
			//Console.WriteLine (" -- AvgTask {0} / {1}, {2} / {3} workers", 
			//    TextHelper.ToStringMS (_PreviousAvgTimeDurationMs), 
			//    TextHelper.ToStringMS (previousAvgTimeDurationMs),
			//    ActiveWorkersCount, 
			//    _Workers.Count);
				
			Console.WriteLine (" -- Adv {0:0.##}%, Tasks {1} / {2}, TotalTime: {3}", 
				previousAdv,
				taskSize, pendingTasksCount,
				TextHelper.ToString (ref _Stopwatch));
				
			Console.WriteLine (" -- Est Total Time: {0}* / {1}", 
				TextHelper.ToStringMS (previousTotalTimeMs),
				TextHelper.ToStringMS (_PreviousTotalTimeMs));
			
			
			// El tiempo pasado era mejor
			if (_PreviousTotalTimeMs > 0 &&
				//_BestEstTotalTimeMs < _PreviousTotalTimeMs)
				previousTotalTimeMs < _PreviousTotalTimeMs)
			{
				// sobran trabajadores
				tmpWorkers = _Workers.FindAll (w => w.Active);
				
				if (tmpWorkers.Count > 0)
				{
					if (_PreviousAddedWorkersCount == 0)
					{
						// duerme al 10% si son los originales
						tmpSleepCount = (int) (tmpWorkers.Count * 0.05);
					}
					else
					{
						// duerme a la mitad de los agregados
						tmpSleepCount = (int) Math.Ceiling (_PreviousAddedWorkersCount / 2.0);
					}
					
					if (ActiveWorkersCount - tmpSleepCount > MIN_WORKERS_COUNT)
					{
						//Console.WriteLine (" -- Durmiendo a {0} workers", tmpSleepCount);
						Parallel.For (0, tmpSleepCount, (i) =>
						{
							tmpWorkers [i].Stop ();
						});
						
						_EnableAddWorkers = false;
					}
				}
			}
			else
			{
				// Se agrega al 50% de los trabajadores necesarios
				if (pendingTasksCount > 0)
				{
					_PreviousAddedWorkersCount += (int) Math.Ceiling (ActiveWorkersCount * 0.05);
					
					AddWorkers (_PreviousAddedWorkersCount);
				}
			}
			
			if (2 * _AvgTimeDurationMs < TimeSpan.MaxValue.Milliseconds 
				&& _AvgTimeDurationMs > CHECKCLEANCYCLEMS)
			{
				Thread.Sleep (2 * _AvgTimeDurationMs);
			}
			else
			{
				Thread.Sleep (CHECKCLEANCYCLEMS);
			}
			
			Clean (pNotUsed);
		}
		
		
		private void ConfirmedTask (int pAvgTimeDurationMs, int pWorkerIndex)
		{
			Interlocked.Decrement (ref _PendingTasksCount);
			
			_AvgTimeDurationMs = pAvgTimeDurationMs;
		}
		
		#endregion
		
		
		#region Manejo de cola
		
		internal bool Dequeue (out ITask pTask, int pWorkerIndex)
		{
			lock (_Tasks)
			{
				for (int i = 0; i < _Tasks.Count; i++)
				{
					lock (_Tasks [i])
					{
						if (_Tasks [i].Count > 0) //
						{
							pTask = _Tasks [i].Dequeue ();
							
							return (pTask != null);
						}
						else
						{
							_Tasks.RemoveAt (i);
							i--;
						}
					}
				}
			}
			
			pTask = null;
			
			return false;
		}
		
		
		public void Enqueue (ITask pTask)
		{
			//while (_PendingTasksCount > TOTAL_MAXLENGTH)
			//{
			//    Thread.Sleep (_AvgTimeDurationMs);
			//}
			
			_EnableAddWorkers = true;
			
			lock (_Tasks)
			{
				if (_Tasks.Count == 0)
				{
					_Tasks.Add (new Queue<ITask> ());
				}
				
				lock (_Tasks [_Tasks.Count - 1])
				{
					if (_Tasks [_Tasks.Count - 1].Count <= INDIVIDUAL_LIST_MAXLENGTH) 
					{
						_Tasks [_Tasks.Count - 1].Enqueue (pTask);
					}
					else
					{
						_Tasks.Add (new Queue<ITask> ());
						_Tasks [_Tasks.Count - 1].Enqueue (pTask);
					}
				}
			}
			
			Interlocked.Increment (ref _PendingTasksCount);
			Interlocked.Increment (ref _TaskSize);
			
			Start ();
		}	
		
		#endregion
	}
}
